/*
 * Copyright (c) 2015. 石头哥哥
 */

package com;


import com.core.ISuWeiService;
import com.utils.FileUtil;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import redis.clients.jedis.*;
import redis.clients.jedis.params.geo.GeoRadiusParam;
import redis.clients.jedis.params.sortedset.ZAddParams;
import redis.clients.jedis.params.sortedset.ZIncrByParams;
import redis.clients.util.SafeEncoder;

import java.io.IOException;
import java.util.*;
import java.util.regex.Pattern;

/**
 * Created by 石头哥哥 on 10/30/15.
 * <p>
 * server-service
 * <p>
 * com.suweia.intellibox.server_service.common.suweia.redis
 * <p>
 * comments:  基于jedis封装操作redis 服务 3314937794 Cleisu13
 */
@SuppressWarnings("Duplicates")
@Component
public class SuWeiJedisService implements ISuWeiService {

    /**
     *
     */
    private Logger logger = LoggerFactory.getLogger(SuWeiJedisService.class);

    /**
     * jedis  pool
     */
    private JedisPool jedisPool;


    /**
     * 集群操作接口
     */
    private JedisCluster jedisCluster;


    /**
     * 一致性hash 集群 操作接口
     */
    private ShardedJedisPool shardedJedisPool;


    /**
     * 集群  节点
     */
    private Set<HostAndPort> hostAndPortSet;


    /**
     * 是否集群   默认采用此方案
     */
    private boolean iscluster;


    /**
     * 开放此对象
     *
     * @return
     */
    public JedisPool getJedisPool() {
        return jedisPool;
    }

    /**
     * 开放此对象
     *
     * @return
     */
    public JedisCluster getJedisCluster() {
        return jedisCluster;
    }

    /**
     *
     */
    public SuWeiJedisService() {
    }

    /**
     * jedis properties 配置文件
     * 默认加载class path
     */
    private String jedisPropertiesPath = "/jedis.properties";

    /**
     * @return
     */
    public String getJedisPropertiesPath() {
        return jedisPropertiesPath;
    }

    public void setJedisPropertiesPath(String jedisPropertiesPath) {
        this.jedisPropertiesPath = jedisPropertiesPath;
    }


    /**
     * 初始化 jedis pool
     * init-method by spring
     * <p>
     * 独立使用: SuWeiJedisService 对象是线程安全 ,可以使用单例模式
     * SuWeiJedisService obj=new SuWeiJedisService();
     * <p>
     * obj.setJedisPropertiesPath("/jedis.properties");
     * <p>
     * obj.initJedisService();
     */
    public void initJedisService() {
        /**
         * loading properties
         */
        Properties properties = FileUtil.buildProperties(jedisPropertiesPath);
        if (properties != null
                && properties.size() != 0) {

            if (logger.isDebugEnabled()) {
                logger.debug("load jedis config success ... ...");
            }

            // is cluster
            iscluster = Boolean.valueOf(properties.getProperty("suweia.cluster"));

            //common config
            GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
            genericObjectPoolConfig.setTestWhileIdle(true);
            genericObjectPoolConfig.setMinEvictableIdleTimeMillis(60000);
            genericObjectPoolConfig.setTimeBetweenEvictionRunsMillis(30000);
            genericObjectPoolConfig.setNumTestsPerEvictionRun(-1);

            genericObjectPoolConfig.setMaxTotal(Integer.valueOf(properties.getProperty("suweia.maxTotal")));
            genericObjectPoolConfig.setMaxIdle(Integer.valueOf(properties.getProperty("suweia.maxIdle")));
            genericObjectPoolConfig.setMinIdle(Integer.valueOf(properties.getProperty("suweia.minIdle")));
            genericObjectPoolConfig.setMaxWaitMillis(Long.valueOf(properties.getProperty("suweia.maxWaitMillis")));
            genericObjectPoolConfig.setTestOnBorrow(Boolean.valueOf(properties.getProperty("suweia.testOnBorrow")));
            genericObjectPoolConfig.setTestOnReturn(Boolean.valueOf(properties.getProperty("suweia.testOnReturn")));


            if (iscluster) {
                //init hosts
                initHostAndPortSet(properties);
                int timeout = Integer.valueOf(properties.getProperty("suwia.timeout"));
                int maxRedirections = Integer.valueOf(properties.getProperty("suwia.maxRedirections"));

                this.jedisCluster = new JedisCluster(this.hostAndPortSet, timeout,
                        maxRedirections, genericObjectPoolConfig);

                if (logger.isDebugEnabled()) {
                    logger.debug("jedisCluster init  success ... ...");
                }
            } else {

                String password = properties.getProperty("suweia.password").equals("") ? null : properties.getProperty("suweia.password");
                this.jedisPool = new JedisPool(genericObjectPoolConfig, properties.getProperty("suweia.host"),
                        Integer.valueOf(properties.getProperty("suweia.port")), Integer.valueOf(properties.getProperty("suweia.timeout"))
                        , password, Integer.valueOf(properties.getProperty("suweia.database"))
                        , properties.getProperty("suweia.clientname"));

                if (logger.isDebugEnabled()) {
                    logger.debug("jedisPool init  success ... ...");
                }
            }

            //clear ref
            properties.clear();
        } else {

            logger.error("\n请创建 :\n" +
                    "jedis.properties,内容为:\n\n" +
                    "##是否使用集群----\n" +
                    "suweia.cluster=true\n" +
                    "##jedis pool 如果不使用集群 则配置\n" +
                    "suweia.host=192.168.0.8\n" +
                    "#suweia.host=127.0.0.1\n" +
                    "suweia.port=6379\n" +
                    "suweia.timeout = 2000\n" +
                    "suweia.password=suwei\n" +
                    "suweia.database=0\n" +
                    "suweia.clientname=suwei_redis\n" +
                    "\n" +
                    "###jedis 集群配置  key :suwia.host_port+number 注意保持唯一性\n" +
                    "### 可做HA\n" +
                    "suwia.host_port1=127.0.0.1:30001\n" +
                    "suwia.host_port2=127.0.0.1:30002\n" +
                    "suwia.host_port3=127.0.0.1:30003\n" +
                    "suwia.host_port4=127.0.0.1:30004\n" +
                    "suwia.host_port5=127.0.0.1:30005\n" +
                    "suwia.host_port6=127.0.0.1:30006\n" +
                    "\n" +
                    "#suwia.host_port1=192.168.0.8:30001\n" +
                    "#suwia.host_port2=192.168.0.8:30002\n" +
                    "#suwia.host_port3=192.168.0.8:30003\n" +
                    "#suwia.host_port4=192.168.0.8:30004\n" +
                    "#suwia.host_port5=192.168.0.8:30005\n" +
                    "#suwia.host_port6=192.168.0.8:30006\n" +
                    "\n" +
                    "#defalut is 2000ms\n" +
                    "suwia.timeout=6000\n" +
                    "suwia.maxRedirections=5\n" +
                    "\n" +
                    "##common GenericObjectPoolConfig\n" +
                    "suweia.maxTotal=2000\n" +
                    "suweia.maxIdle=500\n" +
                    "suweia.maxWaitMillis=  -1\n" +
                    "suweia.testOnBorrow=true\n" +
                    "suweia.testOnReturn=true");
            //exit jvm
            System.exit(0);
        }

//        testJedis();
    }


    /**
     * 初始化 jedisCluter参数
     *
     * @param properties
     */
    @SuppressWarnings("ConstantConditions")
    private void initHostAndPortSet(Properties properties) {
        hostAndPortSet = new HashSet<HostAndPort>();
        Pattern pattern = Pattern.compile("^.+[:]\\d{1,5}\\s*$");//验证配置地址合法性
        for (String config : properties.stringPropertyNames()) {
            if (config.contains("suwia.host_port")) {
                String var = (String) properties.get(config);
                if (pattern.matcher(var).matches()) {
                    String host_port[] = var.split(":");
                    hostAndPortSet.add(new HostAndPort(host_port[0], Integer.valueOf(host_port[1])));
                } else {
                    throw new IllegalArgumentException("ip 或 port 不合法:" + var);
                }
            }
        }
    }

    /**
     * close pool
     */
    public void close() throws IOException {
        if (this.iscluster) {
            this.jedisCluster.close();
        } else {
            this.jedisPool.close();
        }
    }

    /**
     *
     */

    @SuppressWarnings("unchecked")
    @Deprecated
    public void testJedis() {
        try {
            //test jedis


            // test pb
//            User.Builder builder = User.newBuilder();
//            builder.setUsername("石头哥哥");
//            builder.setPwd("123456");
//
////        String pbkey = "pbkey";
//            byte[] user_data = builder.build().toByteArray();
//        String status_code = suwei_set(pbkey, user_data);
//        byte[] user_data_redis = suwei_get(pbkey);
//        User user=null;
//        try {
//            user = User.parseFrom(user_data_redis);
//        } catch (InvalidProtocolBufferException e) {
//            e.printStackTrace();
//        }


            String restaurantId = "restaurantId";

//            for (;;){
//                long status=  suwei_sadd("key",restaurantId.getBytes());
//                long status_2=  suwei_sadd("key","restaurantId".getBytes());
//                System.out.println(status);
//                System.out.println();
//            }


//            suwei_rpush(restaurantId, user_data);
//            suwei_rpush(restaurantId, user_data);
//            suwei_rpush(restaurantId, user_data);
//            suwei_rpush(restaurantId, user_data);

//            ArrayList<byte[]> list = (ArrayList<byte[]>) suwei_lrange(restaurantId, 0, -1);
//            //todo:: test
//            for (byte[] bytes : list) {
//                User.Builder builder1 = User.newBuilder();
//                builder1.mergeFrom(bytes);
//                User _user = builder1.build();
//                System.out.println();
//                //encode data
//            }
//            list.clear();//clear ref
//
//
//            System.out.println();
        } catch (Exception e) {
            e.printStackTrace();

        }

    }


    /**
     * 用于存储 pb
     *
     * @param key
     * @param value
     * @return Status code reply  OK
     */
    @Override
    public String suwei_set(String key, byte[] value) {
        if (iscluster) {
            return jedisCluster.set(key.getBytes(), value);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.set(key.getBytes(), value);
            } finally {
                jedis.close();//return object in jedis pool
            }
        }
    }

    /**
     * 存储序列化后的对象
     * 用于获取pb(定制接口)
     *
     * @param key
     * @return
     */
    @Override
    public byte[] suwei_get(String key) {
        if (iscluster) {
            return jedisCluster.get(key.getBytes());
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.get(key.getBytes());
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 存储对象 ,  设置超时
     * Set the string value as value of the key. The string can't be longer than 1073741824 bytes (1
     * GB).
     *
     * @param key
     * @param value
     * @param time  expire time in the units of <code>expx</code>
     * @return Status code reply
     */
    private static final byte[] NX_BYTES = "NX".getBytes();// NOT  EXIST
    private static final byte[] EX_BYTES = "EX".getBytes();// SECONDS

    @Override
    public String suwei_set(String key, byte[] value, long time) {
        return suwei_set(key.getBytes(), value, NX_BYTES, EX_BYTES, time);
    }

    /**
     * 存储对象 ,  设置超时
     * Set the string value as value of the key. The string can't be longer than 1073741824 bytes (1
     * GB).
     *
     * @param key
     * @param value
     * @param nxxx  NX|XX, NX -- Only set the key if it does not already exist. XX -- Only set the key
     *              if it already exist.
     * @param expx  EX|PX, expire time units: EX = seconds; PX = milliseconds   default is ex
     * @param time  expire time in the units of <code>expx</code>
     * @return Status code reply
     */
    @Override
    public String suwei_set(byte[] key, byte[] value, byte[] nxxx, byte[] expx, long time) {
        if (iscluster) {
            return jedisCluster.set(key, value, nxxx, expx, time);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.set(key, value, nxxx, expx, time);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * Add the specified member to the set value stored at key. If member is already a member of the
     * set no operation is performed. If key does not exist a new set with the specified member as
     * sole member is created. If the key exists but does not hold a set value an error is returned.
     * <p>
     * Time complexity O(1)
     *
     * @param key
     * @param value
     * @return Integer reply, specifically: 1 if the new element was added 0 if the element was
     * already a member of the set
     */
    @Override
    public Long suwei_sadd(String key, byte[] value) {
        if (iscluster) {
            return jedisCluster.sadd(key.getBytes(), value);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.sadd(key.getBytes(), value);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 获取 字节数组
     *
     * @param key
     * @return
     */
    @Override
    public Set<byte[]> suwei_smembers(String key) {
        if (iscluster) {
            return jedisCluster.smembers(key.getBytes());
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.smembers(key.getBytes());
            } finally {
                jedis.close();
            }
        }
    }


    /**
     * 添加到队列尾部
     *
     * @param key
     * @param value
     * @return 返回当前队列中的元素个数
     */
    @Override
    public Long suwei_rpush(String key, byte[] value) {
        if (iscluster) {
            return jedisCluster.rpush(key.getBytes(), value);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.rpush(key.getBytes(), value);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * @param key
     * @return 返回当前队列中的元素个数
     */
    @Override
    public Long suwei_llen(String key) {
        if (iscluster) {
            return jedisCluster.llen(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.llen(key);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 获取指定区间元素 字节数组
     *
     * @param key
     * @param start 0
     * @param end   -1  获取所有元素
     * @return 返回元素list
     */
    @Override
    public List<byte[]> suwei_lrange(final String key, final long start, final long end) {
        if (iscluster) {
            return jedisCluster.lrange(key.getBytes(), start, end);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.lrange(key.getBytes(), start, end);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 删除 start ---end之外的元素
     *
     * @param key
     * @param start
     * @param end
     * @return status code  ok?!
     */
    @Override
    public String suwei_ltrim(final String key, final long start, final long end) {
        if (iscluster) {
            return jedisCluster.ltrim(key, start, end);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.ltrim(key, start, end);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * @param key
     * @return
     */
    @Override
    public Set<String> smembers(String key) {
        if (iscluster) {
            return jedisCluster.smembers(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.smembers(key);
            } finally {
                jedis.close();
            }
        }
    }

    @Override
    public Boolean sismember(String key, String member) {
        if (iscluster) {
            return jedisCluster.sismember(key, member);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.sismember(key, member);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * Set the string value as value of the key.
     * The string can't be longer than 1073741824 bytes (1
     * GB).
     * <p>
     * Time complexity: O(1)
     *
     * @param key
     * @param value
     * @return Status code reply
     */
    @SuppressWarnings("Duplicates")
    @Override
    public String set(String key, String value) {
        if (iscluster) {
            return jedisCluster.set(key, value);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.set(key, value);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * SETNX actually means "SET if Not eXists".
     *
     * @param key
     * @param value
     * @return
     */
    @Override
    public Long setnx(String key, String value) {
        if (iscluster) {
            return jedisCluster.setnx(key, value);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.setnx(key, value);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * Set the string value as value of the key.
     * The string can't be longer than 1073741824 bytes (1
     * GB).
     *
     * @param key
     * @param value
     * @param nxxx  NX|XX, NX -- Only set the key if it does not already exist.
     *              XX -- Only set the key
     *              if it already exist.
     * @param expx  EX|PX, expire time units: EX = seconds; PX = milliseconds
     * @param time  expire time in the units of {@param #expx}
     * @return Status code reply
     */
    @SuppressWarnings("Duplicates")
    @Override
    public String set(String key, String value, String nxxx, String expx, long time) {
        if (iscluster) {
            return jedisCluster.set(key, value, nxxx, expx, time);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.set(key, value, nxxx, expx, time);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * Remove the specified member from the set value stored at key. If member was not a member of the
     * set no operation is performed. If key does not hold a set value an error is returned.
     * <p>
     * Time complexity O(1)
     *
     * @param key     the key of the set
     * @param members the set member to remove
     * @return Integer reply, specifically: 1 if the new element was removed 0 if the new element was
     * not a member of the set
     */
    public Long srem(String key, String... members) {
        if (iscluster) {
            return jedisCluster.srem(key, members);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.srem(key, members);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * Remove the specified member from the set value stored at key. If member was not a member of the
     * set no operation is performed. If key does not hold a set value an error is returned.
     * <p>
     * Time complexity O(1)
     *
     * @param key     the key of the set
     * @param members the set member to remove
     * @return Integer reply, specifically: 1 if the new element was removed 0 if the new element was
     * not a member of the set
     */
    @Override
    public Long suwei_srem(String key, byte[]... members) {
        if (iscluster) {
            return jedisCluster.srem(key.getBytes(), members);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.srem(key.getBytes(), members);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 添加一个元素到set集合中
     * Add the specified member to the set value stored at key. If member is already a member of the
     * set no operation is performed. If key does not exist a new set with the specified member as
     * sole member is created. If the key exists but does not hold a set value an error is returned.
     * <p>
     * Time complexity O(1)
     *
     * @param key
     * @param members
     * @return Integer reply, specifically: 1 if the new element was added 0 if the element was
     * already a member of the set
     */
    @Override
    public Long sadd(String key, String... members) {
        if (iscluster) {
            return jedisCluster.sadd(key, members);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.sadd(key, members);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * The command is exactly equivalent to the following group of commands:
     * {@link #set(String, String) SET} + {@link #expire(String, int) EXPIRE}. The operation is
     * atomic.
     * <p>
     * Time complexity: O(1)
     *
     * @param key
     * @param seconds
     * @param value
     * @return Status code reply
     */
    @Override
    public String setex(String key, int seconds, String value) {
        if (iscluster) {
            return jedisCluster.setex(key, seconds, value);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.setex(key, seconds, value);
            } finally {
                jedis.close();//returnResourceObject
            }
        }
    }

    /**
     * Get the value of the specified key. If the key does not exist null is returned. If the value
     * stored at key is not a string an error is returned because GET can only handle string values.
     * <p>
     * Time complexity: O(1)
     *
     * @param key
     * @return Bulk reply
     */
    @Override
    public String get(String key) {
        if (iscluster) {
            return jedisCluster.get(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.get(key);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * Test if the specified key exists. The command returns "1" if the key exists, otherwise "0" is
     * returned. Note that even keys set with an empty string as value will return "1". Time
     * complexity: O(1)
     *
     * @param key
     * @return Boolean reply, true if the key exists, otherwise false
     */
    @Override
    public Boolean exists(String key) {
        if (iscluster) {
            return jedisCluster.exists(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.exists(key);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * Return the type of the value stored at key in form of a string. The type can be one of "none",
     * "string", "list", "set". "none" is returned if the key does not exist. Time complexity: O(1)
     *
     * @param key
     * @return Status code reply, specifically: "none" if the key does not exist "string" if the key
     * contains a String value "list" if the key contains a List value "set" if the key
     * contains a Set value "zset" if the key contains a Sorted Set value "hash" if the key
     * contains a Hash value
     */
    @Override
    public String type(String key) {
        if (iscluster) {
            return jedisCluster.type(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.type(key);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * Set a timeout on the specified key. After the timeout the key will be automatically deleted by
     * the server. A key with an associated timeout is said to be volatile in Redis terminology.
     * <p>
     * Voltile keys are stored on disk like the other keys, the timeout is persistent too like all the
     * other aspects of the dataset. Saving a dataset containing expires and stopping the server does
     * not stop the flow of time as Redis stores on disk the time when the key will no longer be
     * available as Unix time, and not the remaining seconds.
     * <p>
     * Since Redis 2.1.3 you can update the value of the timeout of a key already having an expire
     * set. It is also possible to undo the expire at all turning the key into a normal key using the
     * {@link #persist(String) PERSIST} command.
     * <p>
     * Time complexity: O(1)
     *
     * @param key
     * @param seconds
     * @return Integer reply, specifically: 1: the timeout was set. 0: the timeout was not set since
     * the key already has an associated timeout (this may happen only in Redis versions <
     * 2.1.3, Redis >= 2.1.3 will happily update the timeout), or the key does not exist.
     * @see <ahref="http://code.google.com/p/redis/wiki/ExpireCommand">ExpireCommand</a>
     */
    @Override
    public Long expire(String key, int seconds) {
        if (iscluster) {
            return jedisCluster.expire(key, seconds);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.expire(key, seconds);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * Undo a {@link #expire(String, int) expire} at turning the expire key into a normal key.
     * <p>
     * Time complexity: O(1)
     *
     * @param key
     * @return Integer reply, specifically: 1: the key is now persist. 0: the key is not persist (only
     * happens when key not set).
     */
    @Override
    public Long persist(String key) {
        if (iscluster) {
            return jedisCluster.persist(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.persist(key);
            } finally {
                jedis.close();//return object
            }
        }
    }

    /**
     * EXPIREAT works exctly like {@link #expire(String, int) EXPIRE} but instead to get the number of
     * seconds representing the Time To Live of the key as a second argument (that is a relative way
     * of specifing the TTL), it takes an absolute one in the form of a UNIX timestamp (Number of
     * seconds elapsed since 1 Gen 1970).
     * <p>
     * EXPIREAT was introduced in order to implement the Append Only File persistence mode so that
     * EXPIRE commands are automatically translated into EXPIREAT commands for the append only file.
     * Of course EXPIREAT can also used by programmers that need a way to simply specify that a given
     * key should expire at a given time in the future.
     * <p>
     * Since Redis 2.1.3 you can update the value of the timeout of a key already having an expire
     * set. It is also possible to undo the expire at all turning the key into a normal key using the
     * {@link #persist(String) PERSIST} command.
     * <p>
     * Time complexity: O(1)
     *
     * @param key
     * @param unixTime
     * @return Integer reply, specifically: 1: the timeout was set. 0: the timeout was not set since
     * the key already has an associated timeout (this may happen only in Redis versions <
     * 2.1.3, Redis >= 2.1.3 will happily update the timeout), or the key does not exist.
     * @see <ahref="http://code.google.com/p/redis/wiki/ExpireCommand">ExpireCommand</a>
     */
    @Override
    public Long expireAt(String key, long unixTime) {
        if (iscluster) {
            return jedisCluster.expireAt(key, unixTime);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.expireAt(key, unixTime);
            } finally {
                jedis.close();//return object
            }
        }
    }

    /**
     * time life
     * <p>
     * The TTL command returns the remaining time to live in seconds of a key that has an
     * {@link #expire(String, int) EXPIRE} set. This introspection capability allows a Redis client to
     * check how many seconds a given key will continue to be part of the dataset.
     *
     * @param key
     * @return Integer reply, returns the remaining time to live in seconds of a key that has an
     * EXPIRE. In Redis 2.6 or older, if the Key does not exists or does not have an
     * associated expire, -1 is returned. In Redis 2.8 or newer, if the Key does not have an
     * associated expire, -1 is returned or if the Key does not exists, -2 is returned.
     */
    @Override
    public Long ttl(String key) {
        if (iscluster) {
            return jedisCluster.ttl(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.ttl(key);
            } finally {
                jedis.close();//return object
            }
        }
    }


    /**
     * @param key
     * @param integer
     * @return
     */
    @Override
    public Long decrBy(String key, long integer) {
        if (iscluster) {
            return jedisCluster.decrBy(key, integer);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.decrBy(key, integer);
            } finally {
                jedis.close();//return object
            }
        }
    }

    /**
     * @param key
     * @return
     */
    @Override
    public Long decr(String key) {
        if (iscluster) {
            return jedisCluster.decr(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.decr(key);
            } finally {
                jedis.close();//return object
            }
        }
    }

    /**
     * @param key
     * @param integer
     * @return
     */
    @Override
    public Long incrBy(String key, long integer) {
        if (iscluster) {
            return jedisCluster.incrBy(key, integer);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.incrBy(key, integer);
            } finally {
                jedis.close();//return object
            }
        }
    }

    /**
     * @param key
     * @return
     */
    @Override
    public Long incr(String key) {
        if (iscluster) {
            return jedisCluster.incr(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.incr(key);
            } finally {
                jedis.close();//return object
            }
        }
    }

    /**
     * If the key already exists and is a string, this command appends the provided value at the end
     * of the string. If the key does not exist it is created and set as an empty string, so APPEND
     * will be very similar to SET in this special case.
     * <p>
     * Time complexity: O(1). The amortized time complexity is O(1) assuming the appended value is
     * small and the already present value is of any size, since the dynamic string library used by
     * Redis will double the free space available on every reallocation.
     *
     * @param key
     * @param value
     * @return Integer reply, specifically the total length of the string after the append operation.
     */
    @Override
    public Long append(String key, String value) {
        if (iscluster) {
            return jedisCluster.append(key, value);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.append(key, value);
            } finally {
                jedis.close();//return object
            }
        }
    }

    /**
     * @param key
     * @param start
     * @param end
     * @return
     */
    @Override
    public String substr(String key, int start, int end) {
        if (iscluster) {
            return jedisCluster.substr(key, start, end);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.substr(key, start, end);
            } finally {
                jedis.close();//return object
            }
        }
    }

    /**
     * cache object attr
     * hash cache
     *
     * @param key
     * @param field
     * @param value
     * @return
     */
    @Override
    public Long hset(String key, String field, String value) {
        if (iscluster) {
            return jedisCluster.hset(key, field, value);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.hset(key, field, value);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * If key holds a hash, retrieve the value associated to the specified field.
     * <p>
     * If the field is not found or the key does not exist, a special 'nil' value is returned.
     * <p>
     * <b>Time complexity:</b> O(1)
     *
     * @param key
     * @param filed
     * @return Bulk reply
     */
    public String hget(String key, String filed) {
        if (iscluster) {
            return jedisCluster.hget(key, filed);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.hget(key, filed);
            } finally {
                jedis.close();
            }
        }

    }

    /**
     * Set the specified hash field to the specified value
     * if the field not exists. <b>Time
     *
     * @param key
     * @param field
     * @param value
     * @return
     */
    @Override
    public Long hsetnx(String key, String field, String value) {
        if (iscluster) {
            return jedisCluster.hsetnx(key, field, value);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.hsetnx(key, field, value);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * Set the respective fields to the respective values. HMSET replaces old values with new values.
     * <p>
     * If key does not exist, a new key holding a hash is created.
     * <p>
     * <b>Time complexity:</b> O(N) (with N being the number of fields)
     *
     * @param key
     * @param hash file and value
     * @return Return OK or Exception if hash is empty
     */
    @Override
    public String hmset(String key, Map<String, String> hash) {
        if (iscluster) {
            String status = jedisCluster.hmset(key, hash);
            if (status.equals("OK")) {
                hash.clear();
            }
            return status;
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                String status = jedis.hmset(key, hash);
                if (status.equals("OK")) {
                    hash.clear();
                }
                return status;
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 批量获取  value
     *
     * @param key
     * @param fields
     * @return value list
     */
    @Override
    public List<String> hmget(String key, String... fields) {
        if (iscluster) {
            return jedisCluster.hmget(key, fields);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.hmget(key, fields);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 获取对应的key所有的属性值
     * Return all the fields and associated values in a hash.
     * fields--value
     *
     * @param key
     * @return
     */
    @Override
    public Map<String, String> hgetAll(String key) {
        if (iscluster) {
            return jedisCluster.hgetAll(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.hgetAll(key);
            } finally {
                jedis.close();//
            }
        }
    }

    @Override
    public void zadd(String key, double score, String member) {
        if (iscluster) {
            jedisCluster.zadd(key, score, member);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                jedis.zadd(key, score, member);
            } finally {
                jedis.close();
            }
        }

    }

    @Override
    public void zadd(String key, double score, String member, ZAddParams params) {
        if (iscluster) {
            jedisCluster.zadd(key, score, member, params);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                jedis.zadd(key, score, member, params);
            } finally {
                jedis.close();
            }
        }
    }


    @Override
    public void zadd(String key, Map<String, Double> scoreMembers) {
        if (iscluster) {
            jedisCluster.zadd(key, scoreMembers);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                jedis.zadd(key, scoreMembers);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 添加成员
     *
     * @param key
     * @param scoreMembers
     * @param params
     */
    @Override
    public void zadd(String key, Map<String, Double> scoreMembers, ZAddParams params) {
        if (iscluster) {
            jedisCluster.zadd(key, scoreMembers, params);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                jedis.zadd(key, scoreMembers, params);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 升序  0,1,2,
     *
     * @param key
     * @param start
     * @param end
     */
    @Override
    public Set<String> zrange(String key, long start, long end) {
        if (iscluster) {
            return jedisCluster.zrange(key, start, end);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.zrange(key, start, end);
            } finally {
                jedis.close();
            }
        }
    }


    /**
     * 删除 指定的成员
     *
     * @param key
     * @param members
     */
    @Override
    public void zrem(String key, String... members) {
        if (iscluster) {
            jedisCluster.zrem(key, members);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                jedis.zrem(key, members);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 升序删除 指定区间元素
     *
     * @param key
     * @param start
     * @param end
     */
    @Override
    public long zremrange(String key, long start, long end) {
        return zrem(key, zrange(key, start, end));
    }

    /**
     * @param key
     * @param remove_set
     * @return
     */
    private long zrem(String key, Set<String> remove_set) {
        int size = remove_set.size();

        byte[][] many = new byte[size][];
        int index = 0;
        for (String member : remove_set) {
            many[index] = SafeEncoder.encode(member);
            ++index;
        }

        remove_set.clear();

        // zrem
        if (iscluster) {
            return jedisCluster.zrem(key.getBytes(), many);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.zrem(key.getBytes(), many);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 倒序删除指定区间元素
     *
     * @param key
     * @param start
     * @param end
     */
    @Override
    public long zrevremrange(String key, long start, long end) {
        return zrem(key, zrevrange(key, start, end));
    }

    @Override
    public void zincrby(String key, double score, String member) {
        if (iscluster) {
            jedisCluster.zincrby(key, score, member);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                jedis.zincrby(key, score, member);
            } finally {
                jedis.close();
            }
        }
    }

    @Override
    public void zincrby(String key, double score, String member, ZIncrByParams params) {
        if (iscluster) {
            jedisCluster.zincrby(key, score, member, params);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                jedis.zincrby(key, score, member, params);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 升序
     * return index
     *
     * @param key
     * @param member
     */
    @Override
    public long zrank(String key, String member) {
        if (iscluster) {
            return jedisCluster.zrank(key, member);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.zrank(key, member);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 降序
     *
     * @param key
     * @param member
     */
    @Override
    public long zrevrank(String key, String member) {
        if (iscluster) {
            return jedisCluster.zrevrank(key, member);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.zrevrank(key, member);
            } finally {
                jedis.close();
            }
        }
    }

    @Override
    public Set<String> zrevrange(String key, long start, long end) {
        if (iscluster) {
            return jedisCluster.zrevrange(key, start, end);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.zrevrange(key, start, end);
            } finally {
                jedis.close();
            }
        }
    }

    @Override
    public Set<Tuple> zrangeWithScores(String key, long start, long end) {
        if (iscluster) {
            return jedisCluster.zrangeWithScores(key, start, end);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.zrangeWithScores(key, start, end);
            } finally {
                jedis.close();
            }
        }
    }

    @Override
    public Set<Tuple> zrevrangeWithScores(String key, long start, long end) {
        if (iscluster) {
            return jedisCluster.zrevrangeWithScores(key, start, end);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.zrevrangeWithScores(key, start, end);
            } finally {
                jedis.close();
            }
        }
    }

    @Override
    public long zcard(String key) {
        if (iscluster) {
            return jedisCluster.zcard(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.zcard(key);
            } finally {
                jedis.close();
            }
        }
    }

    @Override
    public double zscore(String key, String member) {
        if (iscluster) {
            return jedisCluster.zscore(key, member);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.zscore(key, member);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 添加地理信息
     *
     * @param key       位置key
     * @param longitude 经度
     * @param latitude  纬度
     * @param member    位置名称
     * @return add status
     */
    @Override
    public Long geoadd(String key, double longitude, double latitude, String member) {
            if (iscluster) {
            return jedisCluster.geoadd(key, longitude, latitude, member);//cluster
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.geoadd(key, longitude, latitude, member);//
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 添加地理信息
     *
     * @param key
     * @param memberCoordinateMap key is :地理名称 , GeoCoordinate value: 坐标信息(jedis封装经纬度)
     * @return add status
     */
    @Override
    public Long geoadd(String key, Map<String, GeoCoordinate> memberCoordinateMap) {
        if (iscluster) {
            return jedisCluster.geoadd(key, memberCoordinateMap);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.geoadd(key, memberCoordinateMap);
            } finally {
                jedis.close();

            }
        }
    }

    /**
     * 查询指定位置直接的距离
     *
     * @param key     位置信息key
     * @param member1 成员1   (key)
     * @param member2 成员2    (key)
     * @return 返回距离
     */
    @Override
    public Double geodist(String key, String member1, String member2) {
        if (iscluster) {
            return jedisCluster.geodist(key, member1, member2);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.geodist(key, member1, member2);
            } finally {
                jedis.close();

            }
        }
    }

    /**
     * 查询指定位置直接的距离
     *
     * @param key     位置信息key
     * @param member1 成员1   (key)
     * @param member2 成员2    (key)
     * @param unit    {GeoUnit.KM}   距离单位
     * @return
     */
    @Override
    public Double geodist(String key, String member1, String member2, GeoUnit unit) {
        if (iscluster) {
            return jedisCluster.geodist(key, member1, member2, unit);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.geodist(key, member1, member2, unit);
            } finally {
                jedis.close();
            }
        }
    }


    /**
     * 查询位置GEOHASH编码
     *
     * @param key
     * @param members
     * @return
     */
    @Override
    public List<String> geohash(String key, String... members) {
        if (iscluster) {
            return jedisCluster.geohash(key, members);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.geohash(key, members);
            } finally {
                jedis.close();
            }
        }
    }


    /**
     * 查询指定位置的 坐标信息 (经纬度)
     *
     * @param key
     * @param members
     * @return
     */
    @Override
    public List<GeoCoordinate> geopos(String key, String... members) {
        if (iscluster) {
            return jedisCluster.geopos(key, members);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.geopos(key, members);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 查询指定经纬度位置 ----指定半径范围内的 具体对象位置信息
     *
     * @param key       位置key
     * @param longitude 经度
     * @param latitude  纬度
     * @param radius    半径
     * @param unit      单位 KM ,M
     * @return
     */
    @Override
    public List<GeoRadiusResponse> georadius(String key, double longitude, double latitude,
                                             double radius, GeoUnit unit) {
        if (iscluster) {
            return jedisCluster.georadius(key, longitude, latitude, radius, unit);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.georadius(key, longitude, latitude, radius, unit);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 根据经纬度
     * <p>
     * 查询指定经纬度位置 ----指定半径范围内的 具体对象位置信息
     *
     * @param key       位置key
     * @param longitude 经度
     * @param latitude  纬度
     * @param radius    半径
     * @param unit      单位 KM ,M
     * @param param     {@link GeoRadiusParam} 对于查询数据排序方式或者count 等
     * @return 指定位置(半径内) 地理信息数据
     */
    @Override
    public List<GeoRadiusResponse> georadius(String key, double longitude, double latitude,
                                             double radius, GeoUnit unit, GeoRadiusParam param) {
        if (iscluster) {
            return jedisCluster.georadius(key, longitude, latitude, radius, unit, param);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.georadius(key, longitude, latitude, radius, unit, param);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 根据地理位名称来查找----指定半径范围内的 具体对象位置信息
     *
     * @param key    位置key
     * @param member 地理名称
     * @param radius 半径
     * @param unit   单位 KM ,M
     * @return 指定位置(半径内) 地理信息数据
     */
    @Override
    public List<GeoRadiusResponse> georadiusByMember(String key, String member, double radius, GeoUnit unit) {
        if (iscluster) {
            return jedisCluster.georadiusByMember(key, member, radius, unit);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.georadiusByMember(key, member, radius, unit);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * 根据地理位名称来查找----指定半径范围内的 具体对象位置信息
     *
     * @param key    位置key
     * @param member 地理名称
     * @param radius 半径
     * @param unit   单位 KM ,M
     * @param param  {@link GeoRadiusParam} 对于查询数据排序方式或者count 等
     * @return 指定位置(半径内) 地理信息数据
     */
    @Override
    public List<GeoRadiusResponse> georadiusByMember(String key, String member, double radius, GeoUnit
            unit, GeoRadiusParam param) {
        if (iscluster) {
            return jedisCluster.georadiusByMember(key, member, radius, unit, param);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.georadiusByMember(key, member, radius, unit, param);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * @param key
     * @param field
     * @param value
     * @return
     */
    @Override
    public Long hincrBy(String key, String field, long value) {
        if (iscluster) {
            return jedisCluster.hincrBy(key, field, value);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.hincrBy(key, field, value);
            } finally {
                jedis.close();
            }
        }
    }


    /**
     * Test for existence of a specified field in a hash. <b>Time complexity:</b> O(1)
     *
     * @param key
     * @param filed
     * @return Return 1 if the hash stored at key contains the specified field. Return 0 if the key is
     * not found or the field is not present.
     */
    public Boolean hexists(String key, String filed) {
        if (iscluster) {
            return jedisCluster.hexists(key, filed);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.hexists(key, filed);
            } finally {
                jedis.close();
            }
        }

    }

    /**
     * delete filed
     *
     * @param key
     * @param fields
     * @return
     */
    @Override
    public Long hdel(String key, String... fields) {
        if (iscluster) {
            return jedisCluster.hdel(key, fields);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.hdel(key, fields);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * @param key
     * @return
     */
    @Override
    public Long hlen(String key) {
        if (iscluster) {
            return jedisCluster.hlen(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.hlen(key);
            } finally {
                jedis.close();
            }
        }
    }

    /**
     * if has key ,will return >0
     *
     * @param key
     * @return
     */
    @Override
    public Long del(String key) {
        if (iscluster) {
            return jedisCluster.del(key);
        } else {
            Jedis jedis = jedisPool.getResource();//
            try {
                return jedis.del(key);
            } finally {
                jedis.close();//return
            }
        }
    }

    /**
     * if has key ,will return >0
     *
     * @param keys
     * @return
     */
    @SuppressWarnings("Duplicates")
    @Override
    public Long del(String... keys) {
        if (iscluster) {
            return jedisCluster.del(keys);
        } else {
            Jedis jedis = jedisPool.getResource();//
            try {
                return jedis.del(keys);
            } finally {
                jedis.close();//return
            }
        }
    }


    /**
     * file value of set
     *
     * @param key
     * @return Return all the fields in a hash.
     */
    @Override
    public Set<String> hfields(String key) {
        if (iscluster) {
            return jedisCluster.hkeys(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.hkeys(key);
            } finally {
                jedis.close();//
            }
        }
    }

    /**
     * Return all the values in a hash.
     *
     * @param key
     * @return
     */
    @Override
    public List<String> hvals(String key) {
        if (iscluster) {
            return jedisCluster.hvals(key);
        } else {
            Jedis jedis = jedisPool.getResource();
            try {
                return jedis.hvals(key);
            } finally {
                jedis.close();//
            }
        }
    }


}
